#include "CloudMachinesManager.h"
#include "ServerCommon.h"
#include <cstring>
#include <ctime>
#include <cstdio>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <netinet/in.h>
#include <netinet/ip.h>
#include <arpa/inet.h>

CloudMachinesManager::CloudMachinesManager()
{
	ref_cnt++;
	if (ref_cnt == 1) {
		pthread_create(&tidp, NULL, recvMessage, NULL);
		threadEnd = false;
	}
}

CloudMachinesManager::~CloudMachinesManager()
{
	ref_cnt--;
	if (ref_cnt == 0) {
		pthread_mutex_lock(&threadEnd_mutex);
		threadEnd = true;
		pthread_mutex_unlock(&threadEnd_mutex);
		pthread_join(tidp, NULL);
	}
}

std::map<int, struct Machine> CloudMachinesManager::allocMachines(int num)
{
	std::set<int>::iterator iter;
	std::map<int, struct Machine>::iterator res_iter;
	std::map<int, struct Machine> results;
	int i;
	pthread_mutex_lock(&freePool_mutex);
	for (iter=freePool.begin(),i=0;iter!=freePool.end() && i != num;iter++,i++);
	if (i != num) {
		pthread_mutex_unlock(&freePool_mutex);
		return results;
	}
	pthread_mutex_lock(&machineInfo_mutex);
	for (iter=freePool.begin(),i=0;iter!=freePool.end() && i != num;iter++,i++) {
		machineInfo[*iter].end_lock = new pthread_mutex_t;
		pthread_mutex_init(machineInfo[*iter].end_lock, NULL);
		results[*iter] = machineInfo[*iter];
	}
	pthread_mutex_unlock(&machineInfo_mutex);
	freePool.erase(freePool.begin(), iter);
	pthread_mutex_unlock(&freePool_mutex);
	pthread_mutex_lock(&busyPool_mutex);
	for (res_iter=results.begin();res_iter!=results.end();res_iter++)
		busyPool.insert(res_iter->first);
	pthread_mutex_unlock(&busyPool_mutex);
	return results;
}

void CloudMachinesManager::freeMachines(std::map<int, struct Machine> machines)
{
	std::map<int, struct Machine>::iterator iter;
	pthread_mutex_lock(&busyPool_mutex);
	for (iter=machines.begin();iter!=machines.end();iter++) {
		busyPool.erase(iter->first);
		pthread_mutex_destroy(iter->second.end_lock);
		delete iter->second.end_lock;
	}
	pthread_mutex_unlock(&busyPool_mutex);
	pthread_mutex_lock(&freePool_mutex);
	for (iter=machines.begin();iter!=machines.end();iter++)
		freePool.insert(iter->first);
	pthread_mutex_unlock(&freePool_mutex);
	return;
}

std::string CloudMachinesManager::getResult(int num)
{
	std::string tmp;
	pthread_mutex_lock(&results_mutex);
	tmp = results[num];
	pthread_mutex_unlock(&results_mutex);
	return tmp;
}

void* CloudMachinesManager::recvMessage(void *args)
{
	char buf[CMM_MESS_LEN];
	int maxevents = CMM_MAX;
	struct epoll_event *events = new struct epoll_event[CMM_MAX];
	int nfds;
	int mainSocket = socket(PF_INET, SOCK_STREAM, 0);
	struct sockaddr_in addr;
	int clientSocket;
	int i;
	socklen_t addr_len = sizeof(struct sockaddr_in);
	int epoll_fd = epoll_create(CMM_MAX);
	struct epoll_event ev;
	int time_pre = time(NULL);
	bzero(&addr, addr_len);
	addr.sin_family = AF_INET;
	addr.sin_port = htons(CMM_PORT);
	inet_aton(CMM_IP, &addr.sin_addr);
	bind(mainSocket, (struct sockaddr *) &addr, addr_len);
	listen(mainSocket, CMM_MAX);
	ev.events = EPOLLIN;
	ev.data.fd = mainSocket;
	epoll_ctl(epoll_fd, EPOLL_CTL_ADD, mainSocket, &ev);
	while (1) {
		nfds = epoll_wait(epoll_fd, events, maxevents, -1);
		for (i=0;i<nfds;i++) {
			if (events[i].data.fd == mainSocket) {
				clientSocket = accept(mainSocket,
						(struct sockaddr *) &addr,
						&addr_len);
				if (clientSocket > 0) {
					ev.data.fd = clientSocket;
					epoll_ctl(epoll_fd, EPOLL_CTL_ADD,
							clientSocket,&ev);
				// 把这个节点加入Machineinfo和freePool中
					pthread_mutex_lock(&machineInfo_mutex);
					machineInfo[clientSocket].CPUNum = 0;
					machineInfo[clientSocket].CPUPower = 0;
					machineInfo[clientSocket].GPUNum = 0;
					machineInfo[clientSocket].GPUPower = 0;
					machineInfo[clientSocket].alive = time(NULL);
					pthread_mutex_unlock(&machineInfo_mutex);
					pthread_mutex_lock(&freePool_mutex);
					freePool.insert(clientSocket);
					pthread_mutex_unlock(&freePool_mutex);
				}
			} else {
				if (recv(events[i].data.fd, buf, CMM_MESS_LEN, 0) != 80) {
					machineInfo.erase(events[i].data.fd);
					pthread_mutex_lock(&busyPool_mutex);
					busyPool.erase(events[i].data.fd);
					pthread_mutex_unlock(&busyPool_mutex);
					pthread_mutex_lock(&freePool_mutex);
					freePool.erase(events[i].data.fd);
					pthread_mutex_unlock(&freePool_mutex);
					ev.data.fd = events[i].data.fd;
					epoll_ctl(epoll_fd, EPOLL_CTL_DEL,
							events[i].data.fd, &ev);
				} else if (!strncmp(buf, "ping", strlen("ping"))) {
					pthread_mutex_lock(&machineInfo_mutex);
					machineInfo[events[i].data.fd].alive =
						time(NULL);
					pthread_mutex_unlock(&machineInfo_mutex);
				} else if (!strncmp(buf,"end",strlen("end"))) {
					// 把这个节点从busy中去掉,加入free中
					pthread_mutex_lock(&busyPool_mutex);
					busyPool.erase(events[i].data.fd);
					pthread_mutex_unlock(&busyPool_mutex);
					pthread_mutex_lock(&freePool_mutex);
					freePool.insert(events[i].data.fd);
					pthread_mutex_unlock(&freePool_mutex);
				} else if (!strncmp(buf,"data", strlen("data"))) {
					int data_len;
					char *data_buf;
					int j = 0;
					sscanf(buf, "data %d", &data_len);
					data_buf = new char[data_len + 1];
					while (j < data_len) {
						recv(events[i].data.fd, buf, CMM_MESS_LEN, 0);
						if (j + CMM_MESS_LEN >= data_len) {
							for (;j<data_len;j++)
								data_buf[j] = buf[j%CMM_MESS_LEN];
						} else {
							memcpy(data_buf+j, buf, CMM_MESS_LEN);
							j += CMM_MESS_LEN;
						}
					}
					data_buf[data_len] = '\0';
					pthread_mutex_lock(&results_mutex);
					results[events[i].data.fd] = data_buf;
					pthread_mutex_unlock(&results_mutex);
					pthread_mutex_unlock(machineInfo[events[i].data.fd].end_lock);
					delete [] data_buf;
				} else {
					// 把信息加入Machineinfo中
					pthread_mutex_lock(&machineInfo_mutex);
					sscanf(buf, "%d %lf %d %lf %s",
					&machineInfo[events[i].data.fd].CPUNum,
					&machineInfo[events[i].data.fd].CPUPower,
					&machineInfo[events[i].data.fd].GPUNum,
					&machineInfo[events[i].data.fd].GPUPower,
					machineInfo[events[i].data.fd].hostname);
					pthread_mutex_unlock(&machineInfo_mutex);
				}
			}
			if (time(NULL) - time_pre > 300) {
				time_pre = time(NULL);
				std::map<int, struct Machine>::iterator iter;
				pthread_mutex_lock(&machineInfo_mutex);
				for (iter=machineInfo.begin();iter!=machineInfo.end();iter++)
					if (time_pre - iter->second.alive > 600) {
						pthread_mutex_lock(&busyPool_mutex);
						busyPool.erase(iter->first);
						pthread_mutex_unlock(&busyPool_mutex);
						pthread_mutex_lock(&freePool_mutex);
						freePool.insert(iter->first);
						pthread_mutex_unlock(&freePool_mutex);
						machineInfo.erase(iter->first);
					}
				pthread_mutex_unlock(&machineInfo_mutex);
			}
		}
		pthread_mutex_lock(&threadEnd_mutex);
		if (threadEnd) {
			pthread_mutex_unlock(&threadEnd_mutex);
			break;
		}
		pthread_mutex_unlock(&threadEnd_mutex);
	}
	delete [] events;
	return NULL;
}

std::map<int, std::string> CloudMachinesManager::results;
pthread_mutex_t CloudMachinesManager::results_mutex = PTHREAD_MUTEX_INITIALIZER;
std::set<int> CloudMachinesManager::freePool;
pthread_mutex_t CloudMachinesManager::freePool_mutex = PTHREAD_MUTEX_INITIALIZER;
std::set<int> CloudMachinesManager::busyPool;
pthread_mutex_t CloudMachinesManager::busyPool_mutex = PTHREAD_MUTEX_INITIALIZER;
std::map<int, struct Machine> CloudMachinesManager::machineInfo;
pthread_mutex_t CloudMachinesManager::machineInfo_mutex = PTHREAD_MUTEX_INITIALIZER;
bool CloudMachinesManager::threadEnd;
pthread_mutex_t CloudMachinesManager::threadEnd_mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_t CloudMachinesManager::tidp;
int CloudMachinesManager::ref_cnt = 0;
